package com.atguigu.gmall.realtime.common.util;

import com.atguigu.gmall.realtime.common.constant.Constant;

public class SQLUtil {
    public static String getKafkaDDL(String topic , String groupId){
        return "WITH (\n" +
                "  'connector' = 'kafka',\n" +
                "  'topic' = '" + topic + "',\n" +
                "  'properties.bootstrap.servers' = '" + Constant.KAFKA_BROKERS + "',\n" +
                "  'properties.group.id' = '" + groupId +"',\n" +
                "  'scan.startup.mode' = 'latest-offset',\n" +
                "  'format' = 'json'\n" +
                ")";
    }

    public static String getHbaseDDL(String tableName){
        return "WITH (\n" +
                "    'connector' = 'hbase-2.2',\n" +
                "    'table-name' = '" + Constant.HBASE_NAMESPACE + ":" + tableName + "',\n" +
                "    'zookeeper.quorum' = 'hadoop102,hadoop103,hadoop104:2181',\n" +
                "    'lookup.async' = 'true',\n" +
                "    'lookup.cache' = 'PARTIAL',\n" +
                "    'lookup.partial-cache.max-rows' = '200',\n" +
                "    'lookup.partial-cache.expire-after-write' = '1 hour',\n" +
                "    'lookup.partial-cache.expire-after-access' = '1 hour'\n" +
                ")";
    }

    public static String getUpsertKafkaDDL(String topic){
        return "WITH (\n" +
                "    'connector' = 'upsert-kafka',\n" +
                "    'topic' = '" + topic + "',\n" +
                "    'properties.bootstrap.servers' = '" + Constant.KAFKA_BROKERS + "',\n" +
                "    'key.format' = 'json',\n" +
                "    'value.format' = 'json'\n" +
                ")";
    }

    public static String getDorisDDL(String tableName){
        return  " WITH (" +
                "  'connector' = 'doris', " +
                "  'fenodes' = '" + Constant.DORIS_FE_NODES + "', " +
                "  'table.identifier' = '" + Constant.DORIS_DATABASE + "." + tableName + "', " +
                "  'username' = 'root', " +
                "  'password' = 'aaaaaa', " +
                "  'sink.properties.format' = 'json', " +
                "  'sink.enable-2pc' = 'false' " + // 测试阶段可以关闭两阶段提交,方便测试
                ")  ";
    }
}
